package chris.rpc;

import chris.ConFactory;
import com.rabbitmq.client.*;

import java.io.IOException;
import java.nio.charset.Charset;
import java.util.UUID;
import java.util.concurrent.*;

import static chris.rpc.RPCServer.RPC_QUEUE_NAME;

/**
 * RPCClient
 *
 * @author nanxin.chen
 * @since 2017/7/14
 */
public class RPCClient {

    private Connection connection;
    private Channel    channel;
    private String     replyQueueName;

    public RPCClient() throws IOException, TimeoutException {
        ConnectionFactory factory = ConFactory.getConnectionFactory();

        this.connection = factory.newConnection();
        this.channel = connection.createChannel();

        this.replyQueueName = this.channel.queueDeclare().getQueue();
    }

    public String rpcCall(String msg) throws IOException, InterruptedException {
        String corrId = UUID.randomUUID().toString();
        AMQP.BasicProperties props = new AMQP.BasicProperties
                .Builder()
                .correlationId(corrId)
                .replyTo(this.replyQueueName)
                .build();

        this.channel.basicPublish("", RPC_QUEUE_NAME, props, msg.getBytes(Charset.forName("utf-8")));

        final BlockingQueue<String> response = new ArrayBlockingQueue<>(1);

        this.channel.basicConsume(this.replyQueueName, true, new DefaultConsumer(channel) {
            @Override
            public void handleDelivery(String consumerTag, Envelope envelope, AMQP.BasicProperties properties, byte[] body) throws IOException {
                if (properties.getCorrelationId().equals(corrId)) {
                    response.offer(new String(body, "utf-8"));
                }
            }
        });

        return response.take();
    }

    public void close() throws IOException {
        this.connection.close();
    }


    public static void main(String[] args) throws IOException, TimeoutException, InterruptedException {
        ExecutorService executorService = Executors.newFixedThreadPool(10);
        for (int i = 0; i < 10; i++) {
            int fi = i;
            executorService.submit(() -> {
                RPCClient fibonacciRpc = null;
                try {
                    fibonacciRpc = new RPCClient();
                    System.out.println(" [x] Requesting fib(" + fi + ")");
                    String response = fibonacciRpc.rpcCall("" + fi);
                    System.out.println(" [.] " + fi + " Got '" + response + "'");


                } catch (Exception e) {
                    e.printStackTrace();
                } finally {
                    if (fibonacciRpc != null) {
                        try {
                            fibonacciRpc.close();
                        } catch (IOException e) {
                            e.printStackTrace();
                        }
                    }
                }
            });
        }

        executorService.shutdown();
    }
}
